import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class Consumer {
    private String name;
    private Message message;
    public static BlockingQueue<Message> queue = new LinkedBlockingQueue<>();
    public Consumer(String name) {
        this.name = name;
    }

    //判断素数
    private static boolean isPrime(long n) {
        if (n <= 1) return false;
        for (long i = 2; i * i <= n; i++) {
            if (n % i == 0) return false;
        }
        return true;
    }

    public void receiveMessage(Message message) {
        System.out.println(name + " received message: " + message);
    }

    public static void main(String[] args) {

        String rent = new String("rent");
        String opt = new String("opt");

        boolean swt = false;

        //定义线程池，最大线程数为4
        ExecutorService socketThread = Executors.newFixedThreadPool(2);
        int flag = 0;

        while (true) {
            if (flag == 2){
                swt = !swt;
                flag = 0;
            }
            Runnable runnable1 = () -> {//广播模式
                try {
                    String host = "localhost";
                    int port = 8886;

                    Socket socket = new Socket(host, port);
                    OutputStream out = socket.getOutputStream();
                    System.out.println("广播模式！");


                    socket.getOutputStream().write(rent.getBytes());
                    socket.shutdownOutput();

                    InputStream in = socket.getInputStream();
                    ObjectInputStream ois = new ObjectInputStream(in);
                    Message message = (Message) ois.readObject();
                    queue.add(message);
                    System.out.println("Message received: " + message);
                    long num;
                    num = Long.parseLong(message.getBody());
                    if(isPrime(num)){
                        System.out.println("Prime number: " + num);
                    }
                    byte[] buffer = new byte[1024];
                    int len;
                    while ((len = in.read(buffer)) != -1) {
                    }
                    in.close();
                    out.close();
                    socket.close();
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            };
            Runnable runnable2 = () -> {//集群模式
                try {
                    String host = "localhost";
                    int port = 8886;

                    Socket socket = new Socket(host, port);
                    OutputStream out = socket.getOutputStream();
                    System.out.println("集群模式！");


                    socket.getOutputStream().write(opt.getBytes());
                    socket.shutdownOutput();

                    InputStream in = socket.getInputStream();
                    ObjectInputStream ois = new ObjectInputStream(in);
                    Message message = (Message) ois.readObject();
                    queue.add(message);
                    System.out.println("Message received: " + message);
                    long num2;
                    num2 = Long.parseLong(message.getBody());
                    if(isPrime(num2)){
                        System.out.println("Prime number: " + num2);
                    }
                    byte[] buffer = new byte[1024];
                    int len;
                    while ((len = in.read(buffer)) != -1) {
                    }
                    in.close();
                    out.close();
                    socket.close();
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            };
            if (swt){
                socketThread.submit(runnable1);
            }else{
                socketThread.submit(runnable2);
            }
            flag++;
        }
    }
}
